package io.kiki.sba.registry.server.clientInterface.store;


import io.kiki.sba.registry.common.model.ClientToClientInterfaceServerNodeId;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.store.BaseInfo;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.function.BiConsumer;


public abstract class AbstractDataManager<T extends BaseInfo> implements DataManager<T, String, String> {
    private static final Logger logger = LoggerFactory.getLogger(PublishersImpl.class);


    final ConnectDataIndexer connectDataIndexer = new ConnectDataIndexer(getClass().getName());

    @Autowired
    protected Config config;

    AbstractDataManager() {

    }

    protected Tuple<T, Boolean> addData(T data) {
        return connectDataIndexer.add(data.getClientToClientInterfaceServerNodeId(), DataPos.of(data), () -> addDataToStore(data));
    }

    private Tuple<T, Boolean> addDataToStore(T data) {

        Map<String, T> dataMap = getSlotStore().getOrCreate(data.getDataInfoId());
        // quick path
        final String registerId = data.getRegisterId();
        if (dataMap.putIfAbsent(registerId, data) == null) {
            return new Tuple<>(null, true);
        }
        for (; ; ) {
            final T existing = dataMap.get(registerId);
            if (existing == null) {
                if (dataMap.putIfAbsent(registerId, data) == null) {
                    return new Tuple<>(null, true);
                }
            } else {
                if (!existing.registerVersion().orderThan(data.registerVersion())) {
                    logger.warn("[conflict]{},{},exist={}/{},input={}/{}", data.getDataInfoId(), data.getRegisterId(), existing.registerVersion(), existing.getRegisterTimestamp(), data.registerVersion(), data.getRegisterTimestamp());
                    return new Tuple<>(existing, false);
                }
                if (dataMap.replace(registerId, existing, data)) {
                    return new Tuple<>(existing, true);
                }
            }
        }
    }


    @Override
    public T deleteById(String dataInfoId, String registerId) {
        Map<String, T> map = getSlotStore().get(dataInfoId);
        if (CollectionUtils.isEmpty(map)) {
            logger.warn("Delete but not registered, {}", dataInfoId);
            return null;
        }
        T dataToDelete = map.remove(registerId);

        if (dataToDelete == null) {
            logger.warn("Delete but not registered, {}, {}", dataInfoId, registerId);
        }
        return dataToDelete;
    }

    @Override
    public Map<String, T> deleteByClientToClientInterfaceServerNodeId(ClientToClientInterfaceServerNodeId clientToClientInterfaceServerNodeId) {
        Store<T> store = getSlotStore();
        Map<String, T> map = new HashMap<>(128);
        for (DataPos dataPos : connectDataIndexer.queryByKey(clientToClientInterfaceServerNodeId)) {
            Map<String, T> map1 = store.get(dataPos.getDataInfoId());
            if (CollectionUtils.isEmpty(map1)) {
                continue;
            }
            T data = map1.get(dataPos.getRegisterId());
            if (data == null || !data.getClientToClientInterfaceServerNodeId().equals(clientToClientInterfaceServerNodeId)) {
                continue;
            }
            if (map1.remove(dataPos.getRegisterId(), data)) {
                map.put(data.getRegisterId(), data);
            }
        }
        return map;
    }

    @Override
    public Map<ClientToClientInterfaceServerNodeId, Map<String, T>> deleteByClientToClientInterfaceServerNodeIdSet(Set<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIds) {
        Map<ClientToClientInterfaceServerNodeId, Map<String, T>> map = new HashMap<>(clientToClientInterfaceServerNodeIds.size());
        for (ClientToClientInterfaceServerNodeId clientToClientInterfaceServerNodeId : clientToClientInterfaceServerNodeIds) {
            Map<String, T> map1 = deleteByClientToClientInterfaceServerNodeId(clientToClientInterfaceServerNodeId);
            if (!CollectionUtils.isEmpty(map1)) {
                map.put(clientToClientInterfaceServerNodeId, map1);
            }
        }
        return map;
    }

    @Override
    public void forEach(BiConsumer<String, Map<String, T>> biConsumer) {
        getSlotStore().forEach((String dataInfoId, Map<String, T> map) -> {
            if (!CollectionUtils.isEmpty(map)) {
                biConsumer.accept(dataInfoId, Collections.unmodifiableMap(map));
            }
        });
    }

    @Override
    public List<T> getDataList(String dataInfoId) {
        StringUtil.checkNotBlank(dataInfoId, "dataInfoId");
        Map<String, T> map = getSlotStore().get(dataInfoId);
        if (map == null || map.isEmpty()) {
            return Collections.emptyList();
        }
        return new ArrayList<>(map.values());
    }

    @Override
    public Map<String, Map<String, T>> getDatas() {
        return getSlotStore().copyMap();
    }

    @Override
    public List<T> getDataList() {
        List<T> list = new ArrayList<>(1024);
        getSlotStore().forEach((String dataInfoId, Map<String, T> datas) -> {
            list.addAll(datas.values());
        });
        return list;
    }

    @Override
    public Map<String, T> queryByConnectId(ClientToClientInterfaceServerNodeId clientToClientInterfaceServerNodeId) {
        Map<String, T> map = new HashMap<>(128);
        for (DataPos dataPos : connectDataIndexer.queryByKey(clientToClientInterfaceServerNodeId)) {
            T data = queryById(dataPos.getDataInfoId(), dataPos.getRegisterId());
            if (data != null && data.getClientToClientInterfaceServerNodeId().equals(clientToClientInterfaceServerNodeId)) {
                map.put(data.getRegisterId(), data);
            }
        }
        return map;
    }


    @Override
    public Map<ClientToClientInterfaceServerNodeId, Map<String, T>> queryByConnectIds(Set<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIds) {
        Map<ClientToClientInterfaceServerNodeId, Map<String, T>> map = new HashMap<>(clientToClientInterfaceServerNodeIds.size());
        for (ClientToClientInterfaceServerNodeId clientToClientInterfaceServerNodeId : clientToClientInterfaceServerNodeIds) {
            Map<String, T> map1 = queryByConnectId(clientToClientInterfaceServerNodeId);
            if (!CollectionUtils.isEmpty(map1)) {
                map.put(clientToClientInterfaceServerNodeId, map1);
            }
        }
        return map;
    }

    @Override
    public T queryById(String dataInfoId, String registerId) {
        final Map<String, T> map = getSlotStore().get(dataInfoId);
        return map == null ? null : map.get(registerId);
    }

    @Override
    public Tuple<Long, Long> count() {
        return getSlotStore().count();
    }

    @Override
    public Set<ClientToClientInterfaceServerNodeId> getConnectIds() {
        return connectDataIndexer.getKeys();
    }

    @Override
    public Collection<String> getDataInfoIds() {
        return getSlotStore().getDataInfoIds();
    }


    protected abstract Store<T> getSlotStore();

    class ConnectDataIndexer extends DataIndexer<ClientToClientInterfaceServerNodeId, DataPos> {

        public ConnectDataIndexer(String name) {
            super(name);
        }

        @Override
        protected void dataStoreForEach(BiConsumer<ClientToClientInterfaceServerNodeId, DataPos> biConsumer) {
            Store<T> store = getSlotStore();
            if (store == null) {
                return;
            }
            store.forEach((dataInfoId, map) -> {
                for (T data : map.values()) {
                    biConsumer.accept(data.getClientToClientInterfaceServerNodeId(), DataPos.of(data));
                }
            });
        }
    }
}
